package calvin.study.rabbitmq.config;

import java.io.IOException;
import java.util.HashMap;
import java.util.Map;

import org.apache.commons.lang.StringUtils;

import com.rabbitmq.client.AMQP.BasicProperties;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.DefaultConsumer;
import com.rabbitmq.client.Envelope;

public class RabbitMqConsumer extends RabbitMqBase {
	
	/**
	 * Rabbitmq消费者实例化构造函数
	 * @param rountingKey 路由key
	 * @param queueName 队列名称
	 * @param exchangeName 交换器名称
	 */
	public RabbitMqConsumer(String rountingKey, String queueName, String exchangeName) {
		rabbitmqConfig = new RabbitMqConfig(rountingKey, queueName, exchangeName);
		init();
	}

	@Override
	protected RabbitMqModel initConnection(Connection conn) {
		String queueName = rabbitmqConfig.getQueueName();//队列名
		String exchangeName = rabbitmqConfig.getExchangeName();//交互器名称
		String keys = rabbitmqConfig.getKey();//
		try {
			final Channel channel = conn.createChannel();
			channel.exchangeDeclare(exchangeName, "direct", true); // 使用direct模式
			
			Map<String, Object> arguments = new HashMap<String, Object>();
			arguments.put("x-ha-policy", "all");
			//声明队列
			//durable 是否持久化,true,会存盘,服务器重启后任然存在
			//exclusive 是否排他,true,
			//autoDelete 是否自动删除 true 自动删除前提:至少有一个消费者连接到这个队列,之后所有与这个队列连接的消息都断开时,才会自动删除
			//arguments:
				//TTL(x-message-ttl) 整个queue的消息过期
				//Max length(x-max-length) 整个queue的消息条数限制 超过队列最大限制数后成为死信
				//Dead letter exchange(x-dead-letter-exchange) 死信消息转发交换机(topicExchange)
				
			
			
			channel.queueDeclare(queueName, true, false, false, arguments);
			for(String routingKey : StringUtils.split(keys, ",")){
				if (!StringUtils.isEmpty(routingKey)) { // 消费者需要绑定key
					channel.queueBind(queueName, exchangeName, routingKey);
				}
			}
			
			//消费者回调方法
			DefaultConsumer consumer = 	new DefaultConsumer(channel) {
				@Override
				public void handleDelivery(String consumerTag, Envelope envelope, BasicProperties properties,
						byte[] body) throws IOException {
					
					String msg = new String(body,"utf-8");
					System.out.println("Received message.."+ msg);
					 //交换机
                    String exchange = envelope.getExchange();
                    //路由
                    String routingKey = envelope.getRoutingKey();
                    System.out.println("consumerTag: " + consumerTag);
                    //消息id
                    long deliveryTag = envelope.getDeliveryTag();
                    System.out.println("deliveryTag: " + deliveryTag );
                    //消息内容
                    channel.basicAck(deliveryTag, false);
				}
			};
			
			//该consumer监听目标队列,一旦该队列产生消息后,直接交给consumer处理.
			channel.basicConsume(queueName, true, consumer);
			//RabbitMqModel mqModel = new RabbitMqModel(conn, channel, consumer); // 对于消费者，需要指定第三个参数consumer
			return null;
		} catch (Exception e) {
			System.out.println(e.getMessage());
		}
		return null;
	}

}
